package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.UserRegisterBean;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.AtguiguUtil;
import com.atguigu.edu.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsUserUserRegisterWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserRegisterWindow().init(4005,2,
                "DwsUserUserRegisterWindow",
                Constant.TOPIC_DWD_USER_REGISTER);
    }
    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        stream.map(json -> {
      //      String tsString = JSON.parseObject(json).getString("create_time");
      //              Long ts = AtguiguUtil.dateTimeToTs(tsString);
                    long ts = System.currentTimeMillis();
                    return new UserRegisterBean("","",1l,ts);
        })
               .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserRegisterBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(60))
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserRegisterBean>() {
                    @Override
                    public UserRegisterBean reduce(UserRegisterBean value1, UserRegisterBean value2) throws Exception {
                        value1.setRegisterCt(value1.getRegisterCt() + value2.getRegisterCt());
                        value1.setTs(System.currentTimeMillis());
                        return value1;
                    }
                }
                , new ProcessAllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                    @Override
                    public void process(Context ctx,
                                        Iterable<UserRegisterBean> iterable,
                                        Collector<UserRegisterBean> out) throws Exception {
                        UserRegisterBean bean = iterable.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                }
                  )

                .addSink(FlinkSinkUtil.getClickHouseSink("dws_user_user_register_window",
                        UserRegisterBean.class));
    }
}